package org.tea.infrastructure.ipc;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tea.application.ipc.BaseIpcMessage;
import org.tea.application.service.IPCService;
import redis.clients.jedis.*;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Redis 实现的进程间通信的实现类
 *
 * @author lxh
 */
public class RedisIPCServiceImpl<T>
        implements IPCService<T> {

    private final static Logger log = LoggerFactory.getLogger(RedisIPCServiceImpl.class);

    private final JedisPool jedisPool;

    private final ExecutorService subscribeExecutor;

    public RedisIPCServiceImpl(JedisPool jedisPool,
                               ExecutorService subscribeExecutor) {
        this.jedisPool = jedisPool;
        this.subscribeExecutor = subscribeExecutor;
    }

    @Override
    public boolean publish(String topic, T msg) {
        try (Jedis jedis = jedisPool.getResource()) {
            log.debug("write to [{}] msg: {}", topic, msg);
            jedis.publish(topic, new ObjectMapper().writeValueAsString(msg));
            return true;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public BaseIpcMessage<T> subscribe(String topic) {
        log.info("start subscribe form {}", topic);
        final Lock lock = new ReentrantLock();
        final Condition condition = lock.newCondition();
        try (Jedis jedis = jedisPool.getResource()) {
            lock.lock();
            try {
                final ConditionValueRef<T> valueRef = new ConditionValueRef<>();
                subscribeExecutor.execute(() ->
                        jedis.subscribe(new CrawlRedisPubSub<>(lock, condition, valueRef), topic.getBytes()));
                condition.await();
                log.info("subscribe from {} subscribe msg {}", topic, valueRef.getValue());
                return new RedisIPCMessage<>(BaseIpcMessage.SUCCESS, valueRef.getValue());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
        }
    }

    @Override
    public BaseIpcMessage<T> subscribe(String topic, long maxWaitTimes, TimeUnit unit) {
        log.debug("start subscribe form {}", topic);
        final Lock lock = new ReentrantLock();
        final Condition condition = lock.newCondition();
        try (Jedis jedis = jedisPool.getResource()) {
            lock.lock();
            try {
                final ConditionValueRef<T> valueRef = new ConditionValueRef<>();
                subscribeExecutor.submit(() -> {
                    jedis.subscribe(new CrawlRedisPubSub<T>(lock, condition, valueRef));
                });
                boolean awaited = condition.await(maxWaitTimes, unit);
                if (awaited) {
                    log.info("subscribe from {} subscribe msg {}", topic, valueRef.getValue());
                    return new RedisIPCMessage<>(BaseIpcMessage.SUCCESS, valueRef.getValue());
                }
                log.info("subscribe from {} subscribe TEL", topic);
                return new RedisIPCMessage<>(BaseIpcMessage.FAILED, valueRef.getValue());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
        }
    }
}
